package com.sinozo.data.app.function;

import com.sinozo.data.bean.DwdTradeOrder;
import com.sinozo.data.common.DataConfig;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

/**
 * @author zhaoyb
 * @version 1.0
 * @description 订单水位线
 * @date 2023/6/14 19:28:49
 */
public class TradeOrderTimestampsAndWatermarks implements AssignerWithPeriodicWatermarks<DwdTradeOrder> {

    private static final long MAX_OUT_OF_ORDERLINESS = DataConfig.MAX_OUT_OF_ORDERLINESS;

    private long currentMaxTimestamp;

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - MAX_OUT_OF_ORDERLINESS);
    }

    @Override
    public long extractTimestamp(DwdTradeOrder element, long recordTimestamp) {
        long timestamp =  System.currentTimeMillis();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }
}
